package com.dragon.cloud.zconf.client;

import com.dragon.cloud.zconf.util.FileUtils;
import com.dragon.cloud.zconf.util.ZConfConstant;
import com.google.common.base.Function;
import com.google.common.base.Objects;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.listen.ListenerContainer;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;

public class ZConfNodeImpl implements ZConfNode, Closeable {

    public static final String NULL_VALUE = "";
    private static Logger logger = LoggerFactory.getLogger(ZConfNodeImpl.class);
    private final ListenerContainer<ZConfNodeListener> listeners = new ListenerContainer<ZConfNodeListener>();
    private final AtomicBoolean isConnected = new AtomicBoolean(false);
    private String path;
    private String value;
    private NodeCache cache;
    private AtomicBoolean nodeCacheStarted = new AtomicBoolean(false);
    private CuratorFramework curatorFramework = null;
    private ConnectionStateListener connectionStateListener = new ConnectionStateListener() {
        @Override
        public void stateChanged(CuratorFramework client, ConnectionState newState) {

            if ((newState == ConnectionState.CONNECTED) || (newState == ConnectionState.RECONNECTED)) {
                if (isConnected.compareAndSet(false, true)) {
                    try {
                        sync();
                    } catch (Exception e) {
                        logger.error("Trying to reset cache failed after connection/reconnection", e);
                    }
                }
            } else {
                isConnected.set(false);
            }
            logger.info("path[{}] connection status changed to [{}]", path, newState);
        }
    };

    public ZConfNodeImpl(String path, CuratorFramework client) {
        this.curatorFramework = client;
        this.path = path;
        this.cache = new NodeCache(client, path);
        client.getConnectionStateListenable().addListener(connectionStateListener);
        addListener(this.cache);
        // if the connect is normal. start the node cache.
        if (client.getZookeeperClient().isConnected()) {
            isConnected.set(true);
            startNodeCache();
        }
        // set the value.
        setValue();
    }

    @Override
    public String getAppId() {
        if (StringUtils.isNotEmpty(path)) {
            String subPath =
                    path.substring(path.indexOf(ZConfConstant.CONFIG_FILE_PATH) + ZConfConstant.CONFIG_FILE_PATH.length() +
                            1);
            String appId = subPath.substring(0, subPath.indexOf("/"));
            return appId;
        }
        return null;
    }

    @Override
    public String getKey() {
        if (StringUtils.isNotEmpty(path)) {
            String subPath =
                    path.substring(path.indexOf(ZConfConstant.CONFIG_FILE_PATH) + ZConfConstant.CONFIG_FILE_PATH.length() +
                            1);
            String key = subPath.substring(subPath.indexOf("/") + 1);
            return key;
        }
        return null;
    }

    @Override
    public String getPath() {
        return this.path;
    }

    @Override
    public String getValue() {
        if (!nodeCacheStarted.get() && curatorFramework.getZookeeperClient().isConnected()) {
            sync();
        }
        return this.value;
    }

    @Override
    public boolean setValue(String value) {
        if (value == null) {
            logger.warn("illegal value, can not be null. ");
            return false;
        }
        synchronized (this) {
            try {
                curatorFramework.setData().forPath(path, value.getBytes());
                changeValue(value);
                return true;
            } catch (Exception e) {
                logger.warn("set value failed, try to create a node. ", e);
                try {
                    curatorFramework.create().forPath(path, value.getBytes());
                    return true;
                } catch (Exception ex) {
                    logger.error("create value failed.", ex);
                }
            }
        }
        return false;
    }

    @Override
    public synchronized void sync() {
        if (!this.nodeCacheStarted.get()) {
            startNodeCache();
        } else {
            rebuildNodeCache();
        }
        setValue();
    }

    @Override
    public void monitor(ZConfNodeListener listener) {
        // set the listener
        this.listeners.addListener(listener);
    }

    private synchronized void startNodeCache() {
        // judge cache status
        if (!this.nodeCacheStarted.get() && cache != null) {
            try {
                // start the cache
                this.cache.start(true);
                this.nodeCacheStarted.compareAndSet(false, true);
            } catch (Exception e) {
                logger.error("start node cache failed. path:" + this.getPath(), e);
            }
        }
    }

    private void rebuildNodeCache() {
        try {
            if (null != cache) {
                cache.rebuild();
            }
        } catch (Exception e) {
            logger.error("rebuild node cache failed. path:" + this.getPath(), e);
        }
    }

    /**
     * 给本地@{@value}赋值, 后面的{ZConfClient#getConfigByKey}返回的值就是此value
     */
    private void setValue() {
        if (!this.nodeCacheStarted.get()) {
            logger.warn("node cache not started, path: " + this.path + ". read value form local file.");
            value = FileUtils.loadFromLocalFileCache(path);
            logger.info("path[{}] set value[{}] from file", path, value);
        } else {
            if (null != cache.getCurrentData() && null != cache.getCurrentData().getData()) {
                String nodeValue = new String(cache.getCurrentData().getData());
                if (!nodeValue.equals(value)) {
                    changeValue(nodeValue);
                }
            } else {
                // the node has deleted.so the value is null.
                changeValue(NULL_VALUE);
            }
        }

    }

    private void addListener(final NodeCache cache) {
        // changes
        NodeCacheListener cacheListener = new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                if (cache.getCurrentData() != null && cache.getCurrentData().getData() != null) {
                    changeValue(new String(cache.getCurrentData().getData()));
                } else {
                    // the node has deleted.so the value is null.
                    changeValue(NULL_VALUE);
                }
            }
        };
        cache.getListenable().addListener(cacheListener);
    }

    /**
     * 当value发生改变时,为本地{@value}重新赋值
     */
    private void changeValue(String newValue) {
        final String oldValue = this.value;
        this.value = newValue;
        // put value to local file for backups.
        FileUtils.saveToLocalFileCache(path, value);
        // call back listener
        if (!Objects.equal(oldValue, this.value) && null != listeners && listeners.size() > 0) {
            listeners.forEach(new Function<ZConfNodeListener, Void>() {
                @Override
                public Void apply(ZConfNodeListener listener) {
                    try {
                        listener.nodeChanged(oldValue, value);
                    } catch (Exception e) {
                        logger.error("Calling listener", e);
                    }
                    return null;
                }
            });
        }
    }

    @Override
    public void destroy() {
        try {
            this.close();
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void close() throws IOException {
        if (null != cache) {
            cache.close();
            listeners.clear();
        }
    }

}
